// -*- encoding: utf-8 -*-
/*
    Filename         :database.c
    Description      :operations of database inclue r/w and more
    Time             :2021/05/01 15:37:00
    Author           :LoanCold
*/


#include <stdlib.h>
#include "extmem.h"
#include "util.c"

int writeb(unsigned char *blk, unsigned int addr, Buffer *buf){
    printf("写入第%d块数据\n",addr);
    if (writeBlockToDisk(blk, addr, buf) != 0)
    {
        perror("Writing Block Failed!\n");
        exit(-1);
    }
    // printf("end\n");
    return 0;
}

unsigned char *readb(unsigned int addr, Buffer *buf){
    printf("取出第%d块数据\n",addr);
    unsigned char *blk;
    if ((blk = readBlockFromDisk(addr, buf)) == NULL)
    {
        perror("Reading Block Failed!\n");
        exit(-1);
    }
    return blk;
}

void write_lastblk(int addr, int next_addr, unsigned char* blk, Buffer *buf){
        sprintf(blk + BYTES_TUPLE*MAX_TUPLE_PBLK,"%d",next_addr);
        writeb(blk, addr, buf);
}

void write_lastblk_next(int addr, unsigned char* blk, Buffer *buf){
        write_lastblk(addr, addr+1, blk, buf);
}

/// Description: It is frequently used, so defined as a function.
///              This function is only used to check if blk that stores
///              result have reached to last tuple. If so, move to a new
///              blk. Arguments are all pointers, which means they are 
///              all changed through this func.
/// @arg: p_result:which tuple are we  b_result: blk num of blk_result
/// @returns: new p_result  and new b_result(through point)
void check_result_blk_full(int *p_result, int *b_result, unsigned char** blk_result, Buffer *buf){
    if(*p_result == MAX_TUPLE_PBLK){
        write_lastblk_next(*b_result, *blk_result, buf);
        *b_result += 1;
        *blk_result = getNewBlockInBuffer(buf);
        clear_buffer(*blk_result);
        *p_result = 0;
    }
}

// arguments are all pointers, which means they are all changed through this func.
// return 1 if reaches end
int move_next(int *p, int *b, unsigned char** blk, int b_end, Buffer *buf){
    // R moves to next
    (*p)++;
    // if this blk reaches the end
    if(*p == MAX_TUPLE_PBLK){
        // if this way reaches the end
        if(*b == b_end){
            return 1;
        }
        else{
            (*b)++; // next blk
            *p = 0; // point to start of blk
            freeBlockInBuffer(*blk, buf);
            *blk = readb(*b, buf);// read new blk
        }
    }
    return 0;
}


int copy_rest_relative(int p_current, int b_current, unsigned char *blk_current, int b_end,
                        int p_current_result, int b_current_result, unsigned char *blk_current_result,
                        Buffer *buf){
        
    while(1){
        check_result_blk_full(&p_current_result, &b_current_result, &blk_current_result, buf);
        turple_cpy(get_turple_addr(p_current_result, blk_current_result),
            get_turple_addr(p_current, blk_current));
        p_current_result++;
        if(move_next(&p_current, &b_current, &blk_current, b_end, buf)){
            freeBlockInBuffer(blk_current, buf);
            write_lastblk(b_current_result, 0, blk_current_result, buf);
            return b_current_result;
        }
    }
}

int sub_sort(int size_way, int b_start, int b_result, Buffer *buf){
    unsigned char *tmpbuf[MAX_BUF_NUM];
    int b_result_s = b_result;
    // Relative S

    // Get data blk
    for(int k=0; k< size_way; k++){
        for(int i=0; i< MAX_BUF_NUM; i++){
            tmpbuf[i] = readb(b_start+i+k*MAX_BUF_NUM+1,buf);
        }
        // Insert Sort
        for(int i=1; i< MAX_BUF_NUM*MAX_TUPLE_PBLK; i++){
            unsigned char value[8];
            turple_cpy(value, tuple(i,tmpbuf));
            int j;
            for(j=i-1; j>=0; j--){
                if(is_leq(value, tuple(j,tmpbuf))){
                    turple_cpy(tuple(j+1,tmpbuf), value);
                    break;
                }
                turple_cpy(tuple(j+1,tmpbuf), tuple(j,tmpbuf));
            }
            if(j <0)
                turple_cpy(tuple(0,tmpbuf), value);
        }  
        // Write back blk
        for(int i=0; i< MAX_BUF_NUM; i++){
            write_lastblk_next(b_result_s+i, tmpbuf[i], buf);
        }
        b_result_s += MAX_BUF_NUM;
    }
    return b_result_s;
}


/// Description:Two stage multiway merge sort(desc)
///             The arguments must be initialized out of this function
/// @arg:size_way: how many ways to be merged
///      b_start: start blk num of processed relative
///      p_buf: array of pointers to buf, which shows which "tuple" we are processing
///      b_buf: array of pointers to buf, which shows which "blk" we are processing 
///      buf_relative: array of pointers to blk that contains tuples
///      buf: buffer of user
/// @returns:b_result: last blk that stores result
/// What it represents:
///                b_buf[0]
///                  |
/// way0: [][][][buf_s[0]] ->
///           b_buf[1]
///             |
/// way1: [][buf_s[1]][][] ->    [blk_result]
///            b_buf[n]
///               |
/// wayn: [][][buf_s[n]][] ->
int multiway_mergesort(int size_way, int b_start, int *p_buf, int *b_buf, unsigned char** buf_relative, Buffer *buf){
    unsigned char *blk_result = getNewBlockInBuffer(buf);
    int p_result = 0;
    int b_result_s = b_start + size_way*MAX_BUF_NUM;// start of result

    while(1){
        // find max
        int way_max = 0; // indicates the way of max(way 0...size_way)
        for(int j=1; j<size_way; j++){
            unsigned char* addr_max = get_turple_addr(p_buf[way_max],buf_relative[way_max]);
            unsigned char* addr_j = get_turple_addr(p_buf[j],buf_relative[j]);
            if(!buf_relative[way_max]){
                way_max = j;
                continue;
            }
            if(!buf_relative[j]){
                continue;
            }
            if(is_leq(addr_max, addr_j)){
                way_max = j; // indicates the position of max(way 0...size_way)
            }
        }
        // No more blk: exit
        if(!buf_relative[way_max])
            break;
        // store result
        check_result_blk_full(&p_result, &b_result_s, &blk_result, buf);
        turple_cpy(get_turple_addr(p_result, blk_result),get_turple_addr(p_buf[way_max],buf_relative[way_max]));
        p_result++;
        // way max move next
        if(move_next(&p_buf[way_max], &b_buf[way_max], &buf_relative[way_max],
                     b_start+ (1+way_max)*MAX_BUF_NUM - 1, buf)){
            freeBlockInBuffer(buf_relative[way_max], buf);
            buf_relative[way_max] = 0; // Buf NULLs
        }
    }

    write_lastblk(b_result_s, 0, blk_result, buf);
    printf("last blk(%d)\n",b_result_s);
    return b_result_s;
}



int multiway_sort_merge(int b_start_R, int b_start_S, int b_start_result, Buffer *buf){
    unsigned char* blk_R = readb(b_start_R, buf);
    unsigned char* blk_S = readb(b_start_S, buf);
    unsigned char* blk_result = getNewBlockInBuffer(buf);
    int p_S = 0;
    int p_R = 0;
    int p_result = 0;
    int b_S = b_start_S;
    int b_R = b_start_R;
    int b_result = b_start_result;
    int value_R=0;
    int value_S=0;
    int end = 0; // ture if traverse all data of R or S

    clear_buffer(blk_result);
    while(1){
        value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        value_S = get_value(get_turple_addr(p_S, blk_S), 0);
        while(value_R > value_S){
            if(move_next(&p_R, &b_R, &blk_R, b_start_R + SIZE_R - 1, buf)){
                freeBlockInBuffer(blk_R, buf);
                freeBlockInBuffer(blk_S, buf);
                write_lastblk(b_result, 0, blk_result, buf);
                return b_result;
            }
            value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        }
        if(end == 1)
            break;
        if(end == 0 && value_R == value_S){
            int p_R_tmp = p_R; 
            int b_R_tmp = b_R; 
            unsigned char *blk_R_tmp = readb(b_R_tmp, buf);
            while(end == 0 && value_R == value_S){
                // every result tuple consists of double size of origin
                if(p_result == (MAX_TUPLE_PBLK/2)*2 ){
                    write_lastblk_next(b_result++, blk_result, buf);
                    blk_result = getNewBlockInBuffer(buf);
                    clear_buffer(blk_result);
                    p_result = 0;
                }
                turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
                turple_cpy(get_turple_addr(p_result+1, blk_result), get_turple_addr(p_R_tmp, blk_R_tmp));
                p_result += 2;

                // R moves to next
                if(move_next(&p_R_tmp, &b_R_tmp, &blk_R_tmp, b_start_R + SIZE_R - 1, buf)){
                    freeBlockInBuffer(blk_R_tmp, buf);
                    break; 
                }
                value_R = get_value(get_turple_addr(p_R_tmp, blk_R_tmp), 0);
            }
            freeBlockInBuffer(blk_R_tmp, buf);
        }
        // S moves to next
        if(move_next(&p_S, &b_S, &blk_S, b_start_S + SIZE_S - 1, buf)){
            freeBlockInBuffer(blk_R, buf);
            freeBlockInBuffer(blk_S, buf);
            write_lastblk(b_result, 0, blk_result, buf);
            return b_result;
        }
    }
    return b_result;
}

int multiway_sort_intersection(int b_start_R, int b_start_S, int b_start_result, Buffer *buf){
    unsigned char* blk_R = readb(b_start_R, buf);
    unsigned char* blk_S = readb(b_start_S, buf);
    unsigned char* blk_result = getNewBlockInBuffer(buf);
    int p_S = 0;
    int p_R = 0;
    int p_result = 0;
    int b_S = b_start_S;
    int b_R = b_start_R;
    int b_result = b_start_result;
    int value_R=0;
    int value_S=0;
    int end = 0; // ture if traverse all data of R or S

    clear_buffer(blk_result);
    while(1){
        value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        value_S = get_value(get_turple_addr(p_S, blk_S), 0);
        // R moves to next
        while(value_R > value_S){
            if(move_next(&p_R, &b_R, &blk_R, b_start_R + SIZE_R - 1, buf)){
                freeBlockInBuffer(blk_R, buf);
                freeBlockInBuffer(blk_S, buf);
                write_lastblk(b_result, 0, blk_result, buf);
                return b_result;
            }
            value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        }
        if(end == 1)
            break;
        // scan if match
        if(end == 0 && value_R == value_S){
            int p_R_tmp = p_R; 
            int b_R_tmp = b_R; 
            unsigned char *blk_R_tmp = readb(b_R_tmp, buf);
            while(end == 0 && value_R == value_S){
                // every result tuple consists of double size of origin
                if(is_turple_equal(get_turple_addr(p_R_tmp, blk_R_tmp), get_turple_addr(p_S, blk_S))){
                    check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                    turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_R_tmp, blk_R_tmp));
                    p_result ++;
                }
                // R moves to next
                if(move_next(&p_R_tmp, &b_R_tmp, &blk_R_tmp, b_start_R + SIZE_R - 1, buf)){
                    freeBlockInBuffer(blk_R_tmp, buf);
                    break; // END!
                }
                value_R = get_value(get_turple_addr(p_R_tmp, blk_R_tmp), 0);
            }
            freeBlockInBuffer(blk_R_tmp, buf);
        }
        // S moves to next
        if(move_next(&p_S, &b_S, &blk_S, b_start_S + SIZE_S - 1, buf)){
            freeBlockInBuffer(blk_R, buf);
            freeBlockInBuffer(blk_S, buf);
            write_lastblk(b_result, 0, blk_result, buf);
            return b_result;
        }
    }

}

int multiway_sort_minus(int b_start_R, int b_start_S, int b_start_result, Buffer *buf){
    unsigned char* blk_R = readb(b_start_R, buf);
    unsigned char* blk_S = readb(b_start_S, buf);
    unsigned char* blk_result = getNewBlockInBuffer(buf);
    int p_S = 0;
    int p_R = 0;
    int p_result = 0;
    int b_S = b_start_S;
    int b_R = b_start_R;
    int b_result = b_start_result;
    int value_R=0;
    int value_S=0;
    int end = 0; // ture if traverse all data of R or S

    clear_buffer(blk_result);
    while(1){
        value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        value_S = get_value(get_turple_addr(p_S, blk_S), 0);
        // R moves to next
        while(value_R > value_S){
            if(move_next(&p_R, &b_R, &blk_R, b_start_R + SIZE_R - 1, buf)){
                freeBlockInBuffer(blk_R, buf);
                printf("rest S");
                return  copy_rest_relative(p_S, b_S, blk_S,
                         b_start_S+SIZE_S-1,p_result, b_result, blk_result, buf);
            }
            value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        }
        if(end == 0 && value_R == value_S){
            int p_R_tmp = p_R; 
            int b_R_tmp = b_R; 
            unsigned char *blk_R_tmp = readb(b_R_tmp, buf);
            int match = 0;
            while(end == 0 && value_R == value_S){
                if(is_turple_equal(get_turple_addr(p_R_tmp, blk_R_tmp), get_turple_addr(p_S, blk_S))){
                    match = 1;
                    break;
                }
                // R moves to next
                if(move_next(&p_R_tmp, &b_R_tmp, &blk_R_tmp, b_start_R + SIZE_R - 1, buf)){
                    freeBlockInBuffer(blk_R_tmp, buf);
                    break; 
                }
                value_R = get_value(get_turple_addr(p_R_tmp, blk_R_tmp), 0);
            }
            if(!match){
                check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
                p_result ++;
            }
            freeBlockInBuffer(blk_R_tmp, buf);
        }
        else{
            check_result_blk_full(&p_result, &b_result, &blk_result, buf);
            turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
            p_result ++;
        }
        // S moves to next
        if(move_next(&p_S, &b_S, &blk_S, b_start_S + SIZE_S - 1, buf)){
            freeBlockInBuffer(blk_R, buf);
            freeBlockInBuffer(blk_S, buf);
            write_lastblk(b_result, 0, blk_result, buf);
            return b_result;
        }
    }
}

int multiway_sort_union(int b_start_R, int b_start_S, int b_start_result, Buffer *buf){
    unsigned char* blk_R = readb(b_start_R, buf);
    unsigned char* blk_S = readb(b_start_S, buf);
    unsigned char* blk_result = getNewBlockInBuffer(buf);
    int p_S = 0;
    int p_R = 0;
    int p_result = 0;
    int b_S = b_start_S;
    int b_R = b_start_R;
    int b_result = b_start_result;
    int value_R=0;
    int value_S=0;
    int end = 0; // ture if traverse all data of R or S

    clear_buffer(blk_result);
    while(1){
        value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        value_S = get_value(get_turple_addr(p_S, blk_S), 0);
        while(value_R > value_S){
            //write to result
            check_result_blk_full(&p_result, &b_result, &blk_result, buf);
            turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_R, blk_R));
            p_result ++;
            if(move_next(&p_R, &b_R, &blk_R, b_start_R + SIZE_R - 1, buf)){
                freeBlockInBuffer(blk_R, buf);
                return  copy_rest_relative(p_S, b_S, blk_S, b_start_S+SIZE_S-1,p_result, b_result, blk_result, buf);
            }
            value_R = get_value(get_turple_addr(p_R, blk_R), 0);
        }
        if(end == 0 && value_R == value_S){
            int p_R_tmp = p_R; 
            int b_R_tmp = b_R; 
            unsigned char *blk_R_tmp = readb(b_R_tmp, buf);
            int match = 0;
            while(end == 0 && value_R == value_S){
                if(is_turple_equal(get_turple_addr(p_R_tmp, blk_R_tmp), get_turple_addr(p_S, blk_S))){
                    match = 1;
                    break;
                }
                if(move_next(&p_R_tmp, &b_R_tmp, &blk_R_tmp, b_start_R + SIZE_R - 1, buf)){
                    freeBlockInBuffer(blk_R_tmp, buf);
                    break; 
                }
                value_R = get_value(get_turple_addr(p_R_tmp, blk_R_tmp), 0);
            }
            if(!match){
                check_result_blk_full(&p_result, &b_result, &blk_result, buf);
                turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
                p_result ++;
            }
            freeBlockInBuffer(blk_R_tmp, buf);
        }
        else{
            check_result_blk_full(&p_result, &b_result, &blk_result, buf);
            turple_cpy(get_turple_addr(p_result, blk_result), get_turple_addr(p_S, blk_S));
            p_result ++;
        }
        if(move_next(&p_S, &b_S, &blk_S, b_start_S + SIZE_S - 1, buf)){
            freeBlockInBuffer(blk_S, buf);
            return  copy_rest_relative(p_R, b_R, blk_R, b_start_R+SIZE_R-1,p_result, b_result, blk_result, buf);
        }
    }

}